package com.lk.listen;

import com.lk.constant.RabbitmqConst;
import com.lk.constant.RedisConst;
import com.lk.service.UserService;
import com.lk.util.RedisUtil;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.Map;

/**
 * @author : lk
 * @date 2021/11/10 11:31
 * @description : TODO
 * @since 1.0
 **/
@Component
public class UserRabbitMq {


    @Autowired
    private UserService userService;

    @Autowired
    private RedisUtil redisUtil;

    /**
     * 使用 topic(主题交换机)模式
     * key 和queue进行绑定
     * 例如 queue  *.TT.*  key 为a.TT.b 将被绑定到 这个对列上
     *
     * @param message 接受到的数据
     * @param headers 唯一性标注
     * @param channel 通道
     * @throws IOException
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = RabbitmqConst.QUEUE_LOCALHOST_TEST, durable = "false", autoDelete = "true"),
            exchange = @Exchange(value = RabbitmqConst.EXCHANGE_LOCALHOST_TEST, type = ExchangeTypes.TOPIC),
            key = RabbitmqConst.QUEUE_LOCALHOST_TEST), ackMode = "MANUAL")
    public void getMessage(Message message, @Headers Map<String, Object>
            headers, Channel channel) throws IOException {
        MessageProperties messageProperties = message.getMessageProperties();
        // 如果全局ID在redis中存在 则代表消费成功 否则消费失败  解决重复消费的问题
        // 使用redis加上 全局id解决重复消费的问题
        String messageId = messageProperties.getMessageId();
        String id = redisUtil.getMessageId();
        if (StringUtils.isEmpty(id)) {
            String data = new String(message.getBody(), "UTF-8");
            // 手动ack // 解决消息丢失的问题
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            try {
                // 执行的业务逻辑
                userService.getData(data);
                channel.basicAck(deliveryTag, false);
                // 消费成功
                redisUtil.put(messageId);
            } catch (Exception e) {
                // 解决消费失败，一直重试的问题 在配置中配置上重试即可
                throw new RuntimeException("消费消息异常");
            }
        }
    }
}
